package com.yc.knn.group2;

import com.yc.DistanceCalculator;
import com.yc.EuclideanDistanceCalculator;
import com.yc.bean.BankMarketing;
import com.yc.bean.Distance;
import com.yc.bean.Sample;

import java.util.*;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Executor;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadPoolExecutor;

/**
 * knn算法的任务调度类
 */
public class ParalleGroupKnnClassifier {
    private int k; //knn的k，表示近邻几个
    private int numThreads; //线程数
    private boolean parallelSort; //排序是否并行

    private List<BankMarketing> dataSet; //训练集39129条， 拿一条测试数据和它计算距离

    private ThreadPoolExecutor executor; //线程池

    public ParalleGroupKnnClassifier(int k, int numThreads, boolean parallelSort, List<BankMarketing> dataSet) {
        this.k = k;
        this.numThreads = numThreads;
        this.parallelSort = parallelSort;
        this.dataSet = dataSet;
        this.executor = (ThreadPoolExecutor) Executors.newFixedThreadPool((int) (this.numThreads * 1.5));
    }

    public String classify(Sample example) {
        CountDownLatch endController=new CountDownLatch(numThreads); //计算器 14
        Distance[] distances = new Distance[dataSet.size()];
        //1.计算 14个线程中每个线程 它的计算任务的 startIndex，endIndex
        int length = dataSet.size() / numThreads; //分段区间：147
        int startIndex = 0;
        int endIndex = length;
        //List<Thread> list = new ArrayList<>();
        //2.根据numThreads 创建任务，并绑定到线程上
        for (int i = 0; i < numThreads; i++) {
            //计算example这条测试数据与 dataSet中 39129条数据的距离(只要计算 startIndex-endIndex)，将距离的结果存到distances
            GroupDistanceTask task = new GroupDistanceTask(distances, startIndex, endIndex, dataSet, example,new EuclideanDistanceCalculator(),endController);
//            Thread t=new Thread(task);
//            t.start();
//            list.add(t);
            executor.execute(task); //加入线程池后，如何让主程序停下来？

            //计算下一个线程的范围
            startIndex = endIndex;
            // rows1 = 2000
            endIndex = i == numThreads - 2 ? dataSet.size() : endIndex + length;
            System.out.println("第" + i + "个线程的计算范围为:" + startIndex + "-" + endIndex);
        }
        //3.调用每个线程 join()，让主线程停止，号计算时间
//        for(Thread t:list){
//            try {
//                t.join();
//            } catch (InterruptedException e) {
//                e.printStackTrace();
//            }
//        }
        try {
            endController.await();  //相当于上面的join，阻塞，当总共 countdownlatch中的值为0，则解除阻塞
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
        //4.排序距离
        if (parallelSort) {
            //并行排序
            Arrays.parallelSort(distances);
        } else {
            Arrays.sort(distances);
        }
        //5.将前k个样本的标签存到一个Map<String,Integer>
        //                           标签名   次数
        Map<String, Integer> results = new HashMap<>();
        for (int i = 0; i < k; i++) {
            Sample sample = dataSet.get(distances[i].getIndex());
            //sample getTag() ->获取标签
            String tag = sample.getTag();
//            if(results.containsKey(tag)){
//                results.put(tag,results.get(tag)+1);
//            }else{
//                results.put(tag,1);
//            }

            results.merge(tag, 1, (a, b) -> a + b);
        }
        //6.取出map中次数最多的标签名返回
        //传统写法
//        Set<Map.Entry<String,Integer>> set=results.entrySet();
//        int max=0;
//        String maxTag="";
//        for(Map.Entry<String,Integer> entry:set){
//            String tag=entry.getKey();
//            int value=entry.getValue();
//            if(value>max){
//                maxTag=tag;
//            }
//        }
//        return maxTag;
        return Collections.max(results.entrySet(), Map.Entry.comparingByValue()).getKey();
    }

    public void destroy() {
        if (executor != null) {
            executor.shutdown();
        }
    }
}
